package org.apache.flink.web.service;

import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.table.client.gateway.Executor;
import org.apache.flink.web.domain.SqlResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class ClientService {

    @Autowired
    private Executor executor;

    public SqlResult exec(String name, String content, Integer limit) {
        return exec(name, new JobID().toString(), content, limit);
    }

    /**
     * Opens the CLI client for executing SQL statements.
     */
    public SqlResult exec(String name, String sessionId, String content, Integer limit) {
        sessionId = executor.openSession(sessionId);
        Runtime.getRuntime().addShutdownHook(new SqlSession.EmbeddedShutdownThread(sessionId, executor));
        try (SqlSession cli = new SqlSession(sessionId, executor, limit)) {
            executor.setSessionProperty(sessionId, PipelineOptions.NAME.key(), name);
            return cli.execute(content);
        } finally {
            executor.closeSession(sessionId);
        }
    }

}
